{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 12 - Tree Reduce\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to implement a tree reduce in Ray by passing object IDs into remote functions to encode dependencies between tasks.\n",
    "\n",
    "In this exercise, you will use Ray to implement parallel data generation and a parallel tree reduction."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=8, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** These functions will need to be turned into remote functions so that the tree of tasks can be executed in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is a proxy for a function which generates some data.\n",
    "def create_data(i):\n",
    "    time.sleep(0.3)\n",
    "    return i * np.ones(10000)\n",
    "\n",
    "# This is a proxy for an expensive aggregation step (which is also\n",
    "# commutative and associative so it can be used in a tree-reduce).\n",
    "def aggregate_data(x, y):\n",
    "    time.sleep(0.3)\n",
    "    return x * y"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Make the data creation tasks run in parallel. Also aggregate the vectors in parallel. Note that the `aggregate_data` function must be called 7 times. They cannot all run in parallel because some depend on the outputs of others. However, it is possible to first run 4 in parallel, then 2 in parallel, and then 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(1.0)\n",
    "start_time = time.time()\n",
    "\n",
    "# EXERCISE: Here we generate some data. Do this part in parallel.\n",
    "vectors = [create_data(i + 1) for i in range(8)]\n",
    "\n",
    "# Here we aggregate all of the data repeatedly calling aggregate_data. This\n",
    "# can be sped up using Ray.\n",
    "#\n",
    "# NOTE: A direct translation of the code below to use Ray will not result in\n",
    "# a speedup because each function call uses the output of the previous function\n",
    "# call so the function calls must be executed serially.\n",
    "#\n",
    "# EXERCISE: Speed up the aggregation below by using Ray. Note that this will\n",
    "# require restructuring the code to expose more parallelism. First run 4 tasks\n",
    "# aggregating the 8 values in pairs. Then run 2 tasks aggregating the resulting\n",
    "# 4 intermediate values in pairs. then run 1 task aggregating the two resulting\n",
    "# values. Lastly, you will need to call ray.get to retrieve the final result.\n",
    "#\n",
    "# Exposing more parallelism means aggregating the vectors in a DIFFERENT ORDER.\n",
    "# This can be done because we are simply summing the data and the order in\n",
    "# which the values are summed doesn't matter (it's commutative and associative).\n",
    "result = aggregate_data(vectors[0], vectors[1])\n",
    "result = aggregate_data(result, vectors[2])\n",
    "result = aggregate_data(result, vectors[3])\n",
    "result = aggregate_data(result, vectors[4])\n",
    "result = aggregate_data(result, vectors[5])\n",
    "result = aggregate_data(result, vectors[6])\n",
    "result = aggregate_data(result, vectors[7])\n",
    "\n",
    "# NOTE: For clarity, the aggregation above is written out as 7 separate function\n",
    "# calls, but this can be done more easily in a while loop via\n",
    "#\n",
    "#     while len(vectors) > 1:\n",
    "#         vectors = aggregate_data(vectors[0], vectors[1]) + vectors[2:]\n",
    "#     result = vectors[0]\n",
    "#\n",
    "# When expressed this way, the change from serial aggregation to tree-structured\n",
    "# aggregation can be made simply by appending the result of aggregate_data to the\n",
    "# end of the vectors list as opposed to the beginning.\n",
    "#\n",
    "# EXERCISE: Think about why this is true.\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert np.all(result == 40320 * np.ones(10000)), ('Did you remember to '\n",
    "                                                  'call ray.get?')\n",
    "assert duration < 0.3 + 0.9 + 0.3, ('FAILURE: The data generation and '\n",
    "                                    'aggregation took {} seconds. This is '\n",
    "                                    'too slow'.format(duration))\n",
    "assert duration > 0.3 + 0.9, ('FAILURE: The data generation and '\n",
    "                              'aggregation took {} seconds. This is '\n",
    "                              'too fast'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the vectors were aggregated with a tree of tasks.\n",
    "\n",
    "You should be able to see the 8 `create_data` tasks running in parallel followed by 4 `aggregate_data` tasks running in parallel followed by 2 more `aggregate_data` tasks followed by 1 more `aggregate_data` task.\n",
    "\n",
    "In the timeline, click on **View Options** and select **Flow Events** to visualize tasks dependencies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import ray.experimental.ui as ui\n",
    "ui.task_timeline()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
